package com.atguigu.flink.chapter11.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/21 9:59
 */
public class Flink01_Table {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        DataStreamSource<String> stream = env.fromElements("hello atguigu", "atguigu hello 210323", "Hello hello atguigu");
        env.setParallelism(2);
        
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        Table table = tenv.fromDataStream(stream, $("word"));
        tenv.createTemporaryView("t", table);
        
        // 1. table api使用
        // 1.1 内联
        /*table
            .joinLateral(call(Split.class, $("word"))) // 内连接
            //            .leftOuterJoinLateral(call(Split.class, $("word")))
            .select($("word"), $("w"), $("len"))
            .execute()
            .print();*/
        
        // 1.2 注册后使用
        //略
        // 2. sql中使用
        tenv.createTemporaryFunction("my_split", Split.class);
        // as T(w1, len1) 可以省略   T就是给row中的列起别名的
        /*tenv.sqlQuery("select" +
                          " word," +
                          " w1," +
                          " len1 " +
                          "from t " +
                          "join lateral table(my_split(word)) as T(w1, len1) on true")
            .execute()
            .print();*/
        // select   .. from a join b on a.id=b.id
        // select .. from a,b where a.id=b.id
       /* tenv.sqlQuery("select" +
                          " word," +
                          " w1," +
                          " len1 " +
                          "from t, " +
                          " lateral table(my_split(word)) as T(w1, len1)")
            .execute()
            .print();*/
    
        tenv.sqlQuery("select" +
                          " word," +
                          " w1," +
                          " len1 " +
                          "from t " +
                          "left join lateral table(my_split(word)) as T(w1, len1) on true")
            .execute()
            .print();
        
    }
    
    // 泛型: 每行数据的类型
    @FunctionHint(output = @DataTypeHint("ROW<w STRING, len INT>"))
    public static class Split extends TableFunction<Row> {
        public void eval(String s) {
            if (s.length() < 15) {
                return;
            }
            
            String[] data = s.split(" ");
            // 每来一个字符串, for循环执行几次, 就表示这次炸出来了几行
            for (String w : data) {
                collect(Row.of(w, w.length()));
            }
        }
    }
    
}

/*
    原表                          炸裂后的表
"hello atguigu"                     hello           5
                                    atguigu         7
"atguigu hello 210323"              ....

把原表和炸裂后的表进行join得到最终的结果.
    join默认的连接提 原表的每个字段
    
    使用的时候还是需要添加上 on 语法

 */